Add single worker execution mode for native execution #24172

kewang1024 · 2024-11-29T21:01:44Z

Description

== RELEASE NOTES ==

General Changes
* Add single worker mode for native execution :pr:`24172`

tdcmeehan · 2024-12-02T16:26:35Z

Can you help to explain why this feature is or has to be tied to native execution? Perhaps describe the background and motivation?

arhimondr

High level comments

Do we want to support single node execution on per-query basis?

This can be useful to improve latency of tiny queries running on a large cluster. For example a user may know that a query is small and may decide to run it on a multi node cluster in a single node mode.

If decided to support it is necessary to make sure the session property is used consistently through the code.

If decided not to support it for now I think the session property should be removed and a configuration property should only be used.

Should the single node execution mode be native specific?

When running a Java cluster deployment with a dedicated coordinator and a dedicated worker (workers) additional exchanges at worker - coordinator boundary are necessary.

I'm thinking if a simpler mental model would be to always add coordinator-to-worker exchanges when single node execution is requested?

arhimondr · 2024-12-02T16:42:19Z

presto-main/src/main/java/com/facebook/presto/SystemSessionProperties.java

@@ -328,6 +328,7 @@ public final class SystemSessionProperties
    // TODO: Native execution related session properties that are temporarily put here. They will be relocated in the future.
    public static final String NATIVE_AGGREGATION_SPILL_ALL = "native_aggregation_spill_all";
    private static final String NATIVE_EXECUTION_ENABLED = "native_execution_enabled";
+    private static final String NATIVE_SINGLE_WORKER_EXECUTION = "native_single_worker_execution";


nit: Should we stay consistent and use node instead of worker (e.g.: query_max_memory_per_node, force_single_node_output, etc.). Also maybe add the _enabled suffix to make it sound more natural, e.g.: single_node_execution_enabled, isSingleNodeExecutionEnabled(...)

arhimondr · 2024-12-02T16:46:05Z

presto-main/src/main/java/com/facebook/presto/SystemSessionProperties.java

@@ -328,6 +328,7 @@ public final class SystemSessionProperties
    // TODO: Native execution related session properties that are temporarily put here. They will be relocated in the future.
    public static final String NATIVE_AGGREGATION_SPILL_ALL = "native_aggregation_spill_all";
    private static final String NATIVE_EXECUTION_ENABLED = "native_execution_enabled";
+    private static final String NATIVE_SINGLE_WORKER_EXECUTION = "native_single_worker_execution";


Is there a good reason why we want to make it native specific? It feels like this property does not necessarily have to be native specific. However when native execution is enabled the planner may need to add additional exchanges to make it work on native.

arhimondr · 2024-12-02T16:46:59Z

presto-main/src/main/java/com/facebook/presto/sql/analyzer/FeaturesConfig.java

@@ -233,6 +233,7 @@ public class FeaturesConfig
    private boolean pushRemoteExchangeThroughGroupId;
    private boolean isOptimizeMultipleApproxPercentileOnSameFieldEnabled = true;
    private boolean nativeExecutionEnabled;
+    private boolean nativeSingleWorkerExecution;


ditto: same comments as for the session property

arhimondr · 2024-12-02T16:48:25Z

presto-main/src/main/java/com/facebook/presto/sql/planner/PlanFragmenterUtils.java

@@ -94,7 +96,7 @@ public static SubPlan finalizeSubPlan(
            PartitioningHandle partitioningHandle)
    {
        subPlan = reassignPartitioningHandleIfNecessary(metadata, session, subPlan, partitioningHandle);
-        if (!forceSingleNode) {
+        if (!forceSingleNode && !(isNativeExecutionEnabled(session) && isNativeSingleWorkerExecution(session))) {


nit: maybe set the existing forceSingleNode accordingly instead

arhimondr · 2024-12-02T16:53:46Z

...ain/java/com/facebook/presto/sql/planner/optimizations/AddExchangeForNativeSingleWorker.java

+        }
+
+        @Override
+        public PlanNode visitTableFinish(TableFinishNode node, RewriteContext<Void> context)


There are other nodes that need to be run on coordinator:

https://github.com/prestodb/presto/blob/master/presto-main/src/main/java/com/facebook/presto/sql/planner/BasePlanFragmenter.java#L181
https://github.com/prestodb/presto/blob/master/presto-main/src/main/java/com/facebook/presto/sql/planner/BasePlanFragmenter.java#L235

Do we need to add an exchange to support those as well?

arhimondr · 2024-12-02T16:58:21Z

presto-main/src/main/java/com/facebook/presto/sql/planner/PlanOptimizers.java

@@ -813,7 +814,11 @@ public PlanOptimizers(
                        costCalculator,
                        ImmutableSet.of(new ScaledWriterRule())));

-        if (!forceSingleNode) {
+        if (featuresConfig.isNativeExecutionEnabled() && featuresConfig.isNativeSingleWorkerExecution()) {


I wonder if a simpler mental model would be to always add worker to coordinator exchanges when single worker execution is enabled.

This way:

single worker execution can be enabled for normal clusters with more than a single worker and a single coordinator (with schedule on coordinator disabled) on per query basis

For Java execution if coordinator scheduling is enabled an extra exchange is not going to hurt

For example this condition can be kept as isSingleNodeExecutionEnabled(session) and we can call the AddExchangeForNativeSingleWorker as AddWorkerToCoordinatorExchanges

arhimondr · 2024-12-02T16:59:33Z

presto-main/src/main/java/com/facebook/presto/sql/planner/PlanOptimizers.java

@@ -813,7 +814,11 @@ public PlanOptimizers(
                        costCalculator,
                        ImmutableSet.of(new ScaledWriterRule())));

-        if (!forceSingleNode) {
+        if (featuresConfig.isNativeExecutionEnabled() && featuresConfig.isNativeSingleWorkerExecution()) {


I think we need to be consistent on whether session property is supported or not. Otherwise someone may try to enable session property without realizing it not being taking into account everywhere, and end up with a broken plan.

arhimondr · 2024-12-02T17:01:05Z

...ain/java/com/facebook/presto/sql/planner/optimizations/MergeJoinForSortedInputOptimizer.java

@@ -55,7 +57,7 @@ public void setEnabledForTesting(boolean isSet)
    @Override
    public boolean isEnabled(Session session)
    {
-        return isEnabledForTesting || isGroupedExecutionEnabled(session) && preferMergeJoinForSortedInputs(session);
+        return isEnabledForTesting || isGroupedExecutionEnabled(session) && preferMergeJoinForSortedInputs(session) && !(isNativeExecutionEnabled(session) && isNativeSingleWorkerExecution(session));


Maybe let's throw an explicit exception saying that MergeJoin is not supported with single node execution (if requested)

arhimondr · 2024-12-02T17:03:22Z

...rc/test/java/com/facebook/presto/nativeworker/TestPrestoNativeWindowQueriesSingleWorker.java

+
+import java.util.Optional;
+
+public class TestPrestoNativeWindowQueriesSingleWorker


We probably want to test EXPLAIN ANALYZE, metadata delete and select from system tables (as it has a special case for native execution)

arhimondr · 2024-12-02T17:03:58Z

...ution/src/test/java/com/facebook/presto/nativeworker/TestPrestoNativeWriterSingleWorker.java

+    }
+
+    @Override
+    public void testScaleWriters() {


scale writers should have no effect in single node? Do we nee this test?

kaikalur · 2024-12-02T17:33:49Z

High level comments

Do we want to support single node execution on per-query basis?

This can be useful to improve latency of tiny queries running on a large cluster. For example a user may know that a query is small and may decide to run it on a multi node cluster in a single node mode.

We can also potentially use HBO/CBO to decide to run some in single node mode

kewang1024 requested review from a team, jaystarshot, feilong-liu and ClarenceThreepwood as code owners November 29, 2024 21:01

kewang1024 requested review from presto-oss and arhimondr November 29, 2024 21:01

kewang1024 force-pushed the native-single-worker branch from dfa9ee8 to 0d9137e Compare December 2, 2024 07:31

Add single worker execution mode for native execution

d71d256

kewang1024 force-pushed the native-single-worker branch from 0d9137e to d71d256 Compare December 2, 2024 08:45

arhimondr reviewed Dec 2, 2024

View reviewed changes

tdcmeehan self-assigned this Dec 2, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add single worker execution mode for native execution #24172

Add single worker execution mode for native execution #24172

kewang1024 commented Nov 29, 2024 •

edited

Loading

tdcmeehan commented Dec 2, 2024

arhimondr left a comment

arhimondr Dec 2, 2024

arhimondr Dec 2, 2024

arhimondr Dec 2, 2024

arhimondr Dec 2, 2024

arhimondr Dec 2, 2024

arhimondr Dec 2, 2024

arhimondr Dec 2, 2024

arhimondr Dec 2, 2024

arhimondr Dec 2, 2024

arhimondr Dec 2, 2024

kaikalur commented Dec 2, 2024


		import java.util.Optional;

		public class TestPrestoNativeWindowQueriesSingleWorker

Add single worker execution mode for native execution #24172

Are you sure you want to change the base?

Add single worker execution mode for native execution #24172

Conversation

kewang1024 commented Nov 29, 2024 • edited Loading

Description

tdcmeehan commented Dec 2, 2024

arhimondr left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kaikalur commented Dec 2, 2024

kewang1024 commented Nov 29, 2024 •

edited

Loading